package com.disruptor.sample._08complex02.run;

import com.disruptor.sample._08complex02.consumer.Consumer;
import com.disruptor.sample._08complex02.dto.Order;
import com.disruptor.sample._08complex02.producer.Producer;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

public class Complex02Main {

  public static void main(String[] args) throws Exception {
    // 创建 ringBuffer,ProducerType.MULTI 多个生产者
    RingBuffer<Order> ringBuffer =
        RingBuffer.create(ProducerType.MULTI,
            new EventFactory<Order>() {
              @Override
              public Order newInstance() {
                return new Order();
              }
            },
            1024 * 1024,
            new YieldingWaitStrategy());

    SequenceBarrier barriers = ringBuffer.newBarrier();

    /**
     * 三个消费者
     * */
    Consumer[] consumers = new Consumer[3];
    for(int i = 0; i < consumers.length; i++){
      consumers[i] = new Consumer("c" + i);
    }

    /**
     *
     * 将三个消费者添加到 WorkerPool 中*/
    WorkerPool<Order> workerPool =
        new WorkerPool<Order>(ringBuffer,
            barriers,
            new IntEventExceptionHandler(),
            consumers);
    /*
     * workerPool.getWorkerSequences() 把消费者的消费进度和下标和生产者关联
     * 生产者就知道了消费者的消费速率，用来控制产生物品的速率
     * 生成速率是通过栅栏来实现的 SequenceBarrier barriers = ringBuffer.newBarrier();
     * */
    ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
    workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));

    final CountDownLatch latch = new CountDownLatch(1);
    /*
     * 100 个生产者
     */
    for (int i = 0; i < 100; i++) {
      final Producer p = new Producer(ringBuffer);
      new Thread(new Runnable() {
        @Override
        public void run() {
          try {
            latch.await();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          /**
           * 每个生产者都提交 100 个任务
           * */
          for(int j = 0; j < 100; j ++){
            p.onData(UUID.randomUUID().toString());
          }
        }
      }).start();
    }
    // 100个生产者线程都被阻塞，在睡眠2秒之后，让100个线程都开始生产数据
    Thread.sleep(2000);
    System.out.println("---------------开始生产-----------------");
    latch.countDown();
    Thread.sleep(5000);
    System.out.println("总数:" + consumers[0].getCount() );
  }

  static class IntEventExceptionHandler implements ExceptionHandler {
    public void handleEventException(Throwable ex, long sequence, Object event) {}
    public void handleOnStartException(Throwable ex) {}
    public void handleOnShutdownException(Throwable ex) {}
  }

}
